package com.yujiahao.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Stream_Transform {

  def main(args: Array[String]): Unit = {
    //TODO SparkStreaming环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    //这里用的是启动一个客户端进行连接
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 8888)

    //TODO treamform是用于当前采集周期中将RDD 进行转换的操作，是无状态
    // 1、transform可以在任意场景下使用RDD。2、周期性在Driver端执行逻辑。transform的方法叫原语.RDD中的方法叫算子
    dstream.transform(
      rdd => {
        rdd.map(
          num => (num * 2)
        )
      }
    ).print()//注意流式数据一定要对数据进行处理不然就报错
    //TODO 默认情况下，DStream中的数据操作只是针对于当前的计算范围，不保留计算结果，
    //TODO 下一个采集周期的计算等同于重头计算

    //启动采集器
    ssc.start()
    //Driver等待采集器的结束，否则，当前Driver处于阻塞状态
    ssc.awaitTermination()

  }

}
